feat: doubao tts support streaming realtime audio - #2087
Conversation
WalkthroughAdds WebSocket-based streaming support for Volcengine TTS: a new binary message protocol, adaptor changes to route "submit" TTS requests to WS and mark streaming, and a WS response handler that dials, exchanges framed messages, and streams audio chunks back to clients. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant Adapter
participant VolcWS as Volcengine WS
Note right of Adapter #e6f7ff: Adaptor routes "submit" -> streaming
Client->>Adapter: TTS request (operation="submit")
Adapter->>Adapter: save TTS request in context\nset IsStream=true
Adapter->>VolcWS: Dial WS (Authorization header)
VolcWS-->>Adapter: WS connection established
Adapter->>VolcWS: Send framed TTS request (binary)
loop streaming audio
VolcWS-->>Adapter: Message (audio chunk / event)
Adapter-->>Client: Chunked HTTP response (audio bytes)
end
VolcWS-->>Adapter: End event (sequence negative)
Adapter->>Client: Close/finish response
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
relay/channel/volcengine/tts.go (1)
146-155: Close response body on all paths.Defer Close before ReadAll to avoid leaks when ReadAll errors.
Apply this diff:
- body, readErr := io.ReadAll(resp.Body) - if readErr != nil { + defer resp.Body.Close() + body, readErr := io.ReadAll(resp.Body) + if readErr != nil { return nil, types.NewErrorWithStatusCode( errors.New("failed to read volcengine response"), types.ErrorCodeReadResponseBodyFailed, http.StatusInternalServerError, ) } - defer resp.Body.Close()relay/channel/volcengine/adaptor.go (1)
299-305: Fix Authorization header format: use "Bearer " not "Bearer;"The Volcengine Ark /v1/audio/speech API requires the standard format "Authorization: Bearer " with a space. The current code at lines 299-305 incorrectly uses a semicolon:
req.Set("Authorization", "Bearer;"+parts[1])Change to:
req.Set("Authorization", "Bearer "+parts[1])Standard authorization headers use "Bearer token" with a space, not a semicolon. This will cause authentication failures against the Volcengine endpoint.
🧹 Nitpick comments (7)
relay/channel/volcengine/tts.go (3)
216-216: Use request context and a handshake timeout for WS dial.Avoid context.Background(); honor client cancellation and set a sane handshake timeout.
Apply this diff:
+ dialer := *websocket.DefaultDialer + dialer.HandshakeTimeout = 10 * time.Second - conn, resp, dialErr := websocket.DefaultDialer.DialContext(context.Background(), requestURL, header) + conn, resp, dialErr := dialer.DialContext(c.Request.Context(), requestURL, header)And add import:
import ( "context" @@ "github.com/gorilla/websocket" + "time" )
294-302: End-of-stream detection: also honor the negative-seq flag.Guard on MsgTypeFlagNegativeSeq (bitmask) in addition to Sequence<0 to be robust across servers.
Apply this diff:
- if msg.Sequence < 0 { + if (msg.MsgTypeFlag&MsgTypeFlagNegativeSeq) != 0 || msg.Sequence < 0 { c.Status(http.StatusOK) usage = &dto.Usage{ PromptTokens: info.PromptTokens, CompletionTokens: 0, TotalTokens: info.PromptTokens, } return usage, nil }
251-254: Consider disabling proxy buffering for streaming.To improve low-latency playback behind Nginx, add X-Accel-Buffering: no.
Apply this diff:
contentType := getContentTypeByEncoding(encoding) c.Header("Content-Type", contentType) + c.Header("X-Accel-Buffering", "no")relay/channel/volcengine/adaptor.go (1)
59-59: Optional: Avoid generic string keys for Gin context.Use a typed key or a unique package-prefixed string to avoid collisions.
Example:
type ctxKey string const ( contextKeyTTSRequest ctxKey = "volcengine/tts_request" contextKeyResponseFormat ctxKey = "volcengine/response_format" )relay/channel/volcengine/protocols.go (3)
406-411: Simpler trailing-bytes check.Avoid reading an extra byte; just check remaining length.
Apply this diff:
- if _, err := buf.ReadByte(); err != io.EOF { - return fmt.Errorf("unexpected data after message: %v", err) - } + if buf.Len() != 0 { + return fmt.Errorf("unexpected %d bytes after message", buf.Len()) + }
580-593: WaitForEvent should loop until the expected event arrives.Current logic errors on first non-matching message.
Apply this diff:
func WaitForEvent(conn *websocket.Conn, msgType MsgType, eventType EventType) (*Message, error) { for { msg, err := ReceiveMessage(conn) if err != nil { return nil, err } - if msg.MsgType != msgType || msg.EventType != eventType { - return nil, fmt.Errorf("unexpected message: %s", msg) - } - if msg.MsgType == msgType && msg.EventType == eventType { + if msg.MsgType == msgType && msg.EventType == eventType { return msg, nil } + // ignore unrelated messages and continue waiting } }
569-571: Restrict to binary frames or tolerate text gracefully.If the server never sends text, reject text frames; otherwise, ignore them instead of attempting to parse.
Example:
if mt != websocket.BinaryMessage { continue // or: return nil, fmt.Errorf("unexpected message type: %d", mt) }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
relay/channel/volcengine/adaptor.go(6 hunks)relay/channel/volcengine/protocols.go(1 hunks)relay/channel/volcengine/tts.go(3 hunks)
🧰 Additional context used
🧬 Code graph analysis (2)
relay/channel/volcengine/tts.go (4)
relay/common/relay_info.go (1)
RelayInfo(75-122)types/error.go (7)
NewAPIError(87-95)NewErrorWithStatusCode(259-275)ErrorCodeChannelInvalidKey(57-57)ErrorCodeBadResponseStatusCode(70-70)ErrorCodeBadRequestBody(66-66)ErrorCodeBadResponse(71-71)ErrorCode(36-36)relay/channel/volcengine/protocols.go (6)
FullClientRequest(595-607)ReceiveMessage(564-578)MsgType(18-18)MsgTypeError(74-74)MsgTypeFrontEndResultServer(73-73)MsgTypeAudioOnlyServer(72-72)dto/openai_response.go (1)
Usage(222-235)
relay/channel/volcengine/adaptor.go (5)
constant/channel.go (2)
ChannelBaseURLs(60-117)ChannelTypeVolcEngine(45-45)relay/channel/api_request.go (1)
DoApiRequest(61-90)relay/common/relay_info.go (1)
RelayInfo(75-122)types/error.go (3)
NewAPIError(87-95)NewErrorWithStatusCode(259-275)ErrorCodeBadRequestBody(66-66)relay/channel/volcengine/tts.go (1)
VolcengineTTSRequest(21-26)
🔇 Additional comments (1)
relay/channel/volcengine/tts.go (1)
213-217: Fix Authorization header format and verify success code for WebSocket protocol.The Authorization header is missing a space after the semicolon. The spec requires "Bearer; {token}", but line 213 generates
"Bearer;%s"without the space.Additionally, the success code in JSON responses should be 20000000, but line 165 checks for
Code != 3000. However, line 165 is in the HTTP TTS flow, not the WebSocket flow shown at lines 213-217. Verify whether the WebSocket protocol uses a different success code validation mechanism or if HTTP TTS also needs updating to 20000000.
| encoding := mapEncoding(c.GetString(contextKeyResponseFormat)) | ||
| if info.IsStream { | ||
| volcRequestInterface, exists := c.Get(contextKeyTTSRequest) | ||
| if !exists { | ||
| return nil, types.NewErrorWithStatusCode( | ||
| errors.New("volcengine TTS request not found in context"), | ||
| types.ErrorCodeBadRequestBody, | ||
| http.StatusInternalServerError, | ||
| ) | ||
| } | ||
|
|
||
| volcRequest, ok := volcRequestInterface.(VolcengineTTSRequest) | ||
| if !ok { | ||
| return nil, types.NewErrorWithStatusCode( | ||
| errors.New("invalid volcengine TTS request type"), | ||
| types.ErrorCodeBadRequestBody, | ||
| http.StatusInternalServerError, | ||
| ) | ||
| } | ||
|
|
||
| // Get the WebSocket URL | ||
| requestURL, urlErr := a.GetRequestURL(info) | ||
| if urlErr != nil { | ||
| return nil, types.NewErrorWithStatusCode( | ||
| urlErr, | ||
| types.ErrorCodeBadRequestBody, | ||
| http.StatusInternalServerError, | ||
| ) | ||
| } | ||
| return handleTTSWebSocketResponse(c, requestURL, volcRequest, info, encoding) | ||
| } |
There was a problem hiding this comment.
Bug: encoding is re-mapped, breaking "opus" to "mp3".
You store the already-mapped encoding in context, then map it again here. For "ogg_opus" this returns "mp3", causing wrong Content-Type.
Apply this diff:
- encoding := mapEncoding(c.GetString(contextKeyResponseFormat))
+ encoding := c.GetString(contextKeyResponseFormat)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| encoding := mapEncoding(c.GetString(contextKeyResponseFormat)) | |
| if info.IsStream { | |
| volcRequestInterface, exists := c.Get(contextKeyTTSRequest) | |
| if !exists { | |
| return nil, types.NewErrorWithStatusCode( | |
| errors.New("volcengine TTS request not found in context"), | |
| types.ErrorCodeBadRequestBody, | |
| http.StatusInternalServerError, | |
| ) | |
| } | |
| volcRequest, ok := volcRequestInterface.(VolcengineTTSRequest) | |
| if !ok { | |
| return nil, types.NewErrorWithStatusCode( | |
| errors.New("invalid volcengine TTS request type"), | |
| types.ErrorCodeBadRequestBody, | |
| http.StatusInternalServerError, | |
| ) | |
| } | |
| // Get the WebSocket URL | |
| requestURL, urlErr := a.GetRequestURL(info) | |
| if urlErr != nil { | |
| return nil, types.NewErrorWithStatusCode( | |
| urlErr, | |
| types.ErrorCodeBadRequestBody, | |
| http.StatusInternalServerError, | |
| ) | |
| } | |
| return handleTTSWebSocketResponse(c, requestURL, volcRequest, info, encoding) | |
| } | |
| encoding := c.GetString(contextKeyResponseFormat) | |
| if info.IsStream { | |
| volcRequestInterface, exists := c.Get(contextKeyTTSRequest) | |
| if !exists { | |
| return nil, types.NewErrorWithStatusCode( | |
| errors.New("volcengine TTS request not found in context"), | |
| types.ErrorCodeBadRequestBody, | |
| http.StatusInternalServerError, | |
| ) | |
| } | |
| volcRequest, ok := volcRequestInterface.(VolcengineTTSRequest) | |
| if !ok { | |
| return nil, types.NewErrorWithStatusCode( | |
| errors.New("invalid volcengine TTS request type"), | |
| types.ErrorCodeBadRequestBody, | |
| http.StatusInternalServerError, | |
| ) | |
| } | |
| // Get the WebSocket URL | |
| requestURL, urlErr := a.GetRequestURL(info) | |
| if urlErr != nil { | |
| return nil, types.NewErrorWithStatusCode( | |
| urlErr, | |
| types.ErrorCodeBadRequestBody, | |
| http.StatusInternalServerError, | |
| ) | |
| } | |
| return handleTTSWebSocketResponse(c, requestURL, volcRequest, info, encoding) | |
| } |
🤖 Prompt for AI Agents
In relay/channel/volcengine/adaptor.go around lines 356 to 386, the code re-maps
an encoding already stored in the request context which transforms "ogg_opus"
into "mp3" and causes an incorrect Content-Type; fix by not re-mapping the
encoding there — read the encoding value from context (or use the already-mapped
value stored earlier) and pass it through to handleTTSWebSocketResponse
unchanged, or only map if no mapped value exists in context.
| _, err = buf.ReadByte() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
There was a problem hiding this comment.
Unmarshal ignores MsgType and Flags from header.
Parse and set them from the second header byte for robustness.
Apply this diff:
- _, err = buf.ReadByte()
- if err != nil {
- return err
- }
+ typeAndFlag, err := buf.ReadByte()
+ if err != nil {
+ return err
+ }
+ m.MsgType = MsgType(typeAndFlag >> 4)
+ m.MsgTypeFlag = MsgTypeFlagBits(typeAndFlag & 0b00001111)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| _, err = buf.ReadByte() | |
| if err != nil { | |
| return err | |
| } | |
| typeAndFlag, err := buf.ReadByte() | |
| if err != nil { | |
| return err | |
| } | |
| m.MsgType = MsgType(typeAndFlag >> 4) | |
| m.MsgTypeFlag = MsgTypeFlagBits(typeAndFlag & 0b00001111) |
| serializationCompression, err := buf.ReadByte() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| m.Serialization = SerializationBits(serializationCompression & 0b11110000) | ||
| m.Compression = CompressionBits(serializationCompression & 0b00001111) | ||
|
|
There was a problem hiding this comment.
Bug: serialization/compression nibbles not shifted.
High 4 bits must be right-shifted; current code stores 0x10 etc instead of 0x1.
Apply this diff:
- m.Serialization = SerializationBits(serializationCompression & 0b11110000)
- m.Compression = CompressionBits(serializationCompression & 0b00001111)
+ m.Serialization = SerializationBits((serializationCompression >> 4) & 0b00001111)
+ m.Compression = CompressionBits(serializationCompression & 0b00001111)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| serializationCompression, err := buf.ReadByte() | |
| if err != nil { | |
| return err | |
| } | |
| m.Serialization = SerializationBits(serializationCompression & 0b11110000) | |
| m.Compression = CompressionBits(serializationCompression & 0b00001111) | |
| serializationCompression, err := buf.ReadByte() | |
| if err != nil { | |
| return err | |
| } | |
| m.Serialization = SerializationBits((serializationCompression >> 4) & 0b00001111) | |
| m.Compression = CompressionBits(serializationCompression & 0b00001111) | |
🤖 Prompt for AI Agents
In relay/channel/volcengine/protocols.go around lines 379 to 386, the code masks
the high nibble for Serialization but doesn't right-shift it, so values like
0x10 are stored instead of 0x1; change the assignment to right-shift the high 4
bits by 4 when setting m.Serialization and keep the low 4 bits for m.Compression
(i.e., mask 0xF0 then >>4 for Serialization, mask 0x0F for Compression),
ensuring proper casting to the SerializationBits and CompressionBits types.
| func (m *Message) readPayload(buf *bytes.Buffer) error { | ||
| var size uint32 | ||
| if err := binary.Read(buf, binary.BigEndian, &size); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if size > 0 { | ||
| m.Payload = buf.Next(int(size)) | ||
| } | ||
|
|
||
| return nil | ||
| } |
There was a problem hiding this comment.
Validate payload length exactly to catch truncation.
Ensure Next() returns the expected size; otherwise error.
Apply this diff:
- if size > 0 {
- m.Payload = buf.Next(int(size))
- }
+ if size > 0 {
+ m.Payload = buf.Next(int(size))
+ if uint32(len(m.Payload)) != size {
+ return fmt.Errorf("payload truncated: expected %d, got %d", size, len(m.Payload))
+ }
+ }🤖 Prompt for AI Agents
In relay/channel/volcengine/protocols.go around lines 551 to 562, the
readPayload function must validate that the buffer contains exactly the
requested payload length to detect truncation; before calling
buf.Next(int(size)), check that buf.Len() >= int(size) (and optionally that size
is within a sane maximum), then call buf.Next and verify the returned slice
length equals int(size); if not, return a descriptive error indicating truncated
payload. Ensure you still set m.Payload when successful and propagate any read
error.
| c.Header("Content-Type", contentType) | ||
| c.Header("Transfer-Encoding", "chunked") | ||
|
|
There was a problem hiding this comment.
Don’t set Transfer-Encoding manually; let net/http handle chunking.
Setting "Transfer-Encoding: chunked" can conflict with Go’s HTTP/2 and proxy behavior. Remove it and rely on Flush().
Apply this diff:
- c.Header("Transfer-Encoding", "chunked")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| c.Header("Content-Type", contentType) | |
| c.Header("Transfer-Encoding", "chunked") | |
| c.Header("Content-Type", contentType) | |
🤖 Prompt for AI Agents
In relay/channel/volcengine/tts.go around lines 253 to 255, remove the manual
c.Header("Transfer-Encoding", "chunked") call; do not set Transfer-Encoding
yourself because Go's net/http (and HTTP/2/proxies) manages chunking. Keep the
Content-Type header, ensure the response writer implements http.Flusher and you
call Flush() where appropriate to stream data, and delete the Transfer-Encoding
header line so chunking is handled by the net/http server.
| // 0 1 2 3 | ||
| // | 0 1 2 3 4 5 6 7 | 0 1 2 3 4 5 6 7 | 0 1 2 3 4 5 6 7 | 0 1 2 3 4 5 6 7 | | ||
| // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | ||
| // | Version | Header Size | Msg Type | Flags | | ||
| // | (4 bits) | (4 bits) | (4 bits) | (4 bits) | | ||
| // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | ||
| // | Serialization | Compression | Reserved | | ||
| // | (4 bits) | (4 bits) | (8 bits) | | ||
| // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | ||
| // | | | ||
| // | Optional Header Extensions | | ||
| // | (if Header Size > 1) | | ||
| // | | | ||
| // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | ||
| // | | | ||
| // | Payload | | ||
| // | (variable length) | | ||
| // | | | ||
| // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
| m.Serialization = SerializationBits(serializationCompression & 0b11110000) | ||
| m.Compression = CompressionBits(serializationCompression & 0b00001111) |
There was a problem hiding this comment.
m.Serialization 好像没有用上，如果不需要的话可以不解析，如果解析的话还是移位后再存储好一些
| continue | ||
| case MsgTypeAudioOnlyServer: | ||
| if len(msg.Payload) > 0 { | ||
| audioBuffer = append(audioBuffer, msg.Payload...) |
| SerializationThrift SerializationBits = 0b11 | ||
| SerializationCustom SerializationBits = 0b1111 |
There was a problem hiding this comment.
SerializationThrift 和 SerializationCustom 我没有在火山的文档中找到，这个是在哪里定义的
|
重构删除无用代码 |
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (9)
relay/channel/volcengine/adaptor.go (3)
44-101: Critical: Streaming enabled for non-official endpoints.Line 92 sets
info.IsStream = truesolely based onOperation == "submit", but Line 262 only returns a WebSocket URL for the official Volcengine base URL. If a user configures a custom HTTP base URL and submits withoperation: "submit",DoRequest(Line 320) will bypass the HTTP request, andDoResponse(Line 331) will attempt WebSocket operations on an HTTP endpoint, causing failures.Apply this diff to gate streaming by the official endpoint:
c.Set(contextKeyTTSRequest, volcRequest) - if volcRequest.Request.Operation == "submit" { + baseUrl := info.ChannelBaseUrl + if baseUrl == "" { + baseUrl = channelconstant.ChannelBaseURLs[channelconstant.ChannelTypeVolcEngine] + } + if baseUrl == channelconstant.ChannelBaseURLs[channelconstant.ChannelTypeVolcEngine] && + volcRequest.Request.Operation == "submit" { info.IsStream = true }
328-362: Critical: Double-mapping encoding corrupts format.Line 330 calls
mapEncoding(c.GetString(contextKeyResponseFormat))on an already-mapped encoding stored at Line 58. For example, if the user requests"opus", Line 56 maps it to"ogg_opus"and stores it in context. Line 330 then maps"ogg_opus"again, which returns"mp3"(since"ogg_opus"isn't a key inresponseFormatToEncodingMap), causing the wrong Content-Type header.Apply this diff to use the already-mapped encoding:
- encoding := mapEncoding(c.GetString(contextKeyResponseFormat)) + encoding := c.GetString(contextKeyResponseFormat)
330-361: Major: DoResponse should validate base URL for WebSocket.Lines 331-360 attempt WebSocket operations when
info.IsStreamis true, but ifinfo.IsStreamwas incorrectly set (due to the issue at Line 92), this will fail for non-official endpoints. Even after fixing Line 92, adding a defensive check here improves robustness.Apply this diff to add a guard:
encoding := c.GetString(contextKeyResponseFormat) if info.IsStream { + baseUrl := info.ChannelBaseUrl + if baseUrl == "" { + baseUrl = channelconstant.ChannelBaseURLs[channelconstant.ChannelTypeVolcEngine] + } + // Only use WebSocket for official Volcengine endpoint + if baseUrl != channelconstant.ChannelBaseURLs[channelconstant.ChannelTypeVolcEngine] { + return handleTTSResponse(c, resp, info, encoding) + } volcRequestInterface, exists := c.Get(contextKeyTTSRequest)relay/channel/volcengine/protocols.go (5)
317-321: Critical: MsgType and Flags discarded during unmarshal.Line 317 reads the second header byte containing
MsgTypeandMsgTypeFlagbut discards it with_. This contradicts Line 233 inNewMessageFromBytes, which correctly extracts these fields. The result is thatUnmarshalleavesm.MsgTypeandm.MsgTypeFlagat their zero values, breaking message parsing.Apply this diff to parse and store both fields:
- _, err = buf.ReadByte() - if err != nil { - return err - } + typeAndFlag, err := buf.ReadByte() + if err != nil { + return err + } + m.MsgType = MsgType(typeAndFlag >> 4) + m.MsgTypeFlag = MsgTypeFlagBits(typeAndFlag & 0b00001111)
322-329: Critical: Serialization field stores unshifted nibble.Line 327 masks the high nibble (
0b11110000) but doesn't right-shift it, som.Serializationwill hold0x10instead of0x1forSerializationJSON. This breaks serialization type checks and marshaling round-trips.Apply this diff to shift the high nibble into the low 4 bits:
- m.Serialization = SerializationBits(serializationCompression & 0b11110000) - m.Compression = CompressionBits(serializationCompression & 0b00001111) + m.Serialization = SerializationBits((serializationCompression >> 4) & 0b00001111) + m.Compression = CompressionBits(serializationCompression & 0b00001111)
356-374: Critical: Flag checks use equality; combined flags fail.Lines 357, 363 use equality (
==) to testMsgTypeFlagWithEventand sequence flags. When flags are combined (e.g., event + positive sequence =0b101), equality checks fail and necessary fields are omitted from marshaling. The same issue affects Lines 422-440 inreaders().Apply this diff to use bitmask checks:
- if m.MsgTypeFlag == MsgTypeFlagWithEvent { + if (m.MsgTypeFlag & MsgTypeFlagWithEvent) != 0 { writers = append(writers, m.writeEvent, m.writeSessionID) } @@ - if m.MsgTypeFlag == MsgTypeFlagPositiveSeq || m.MsgTypeFlag == MsgTypeFlagNegativeSeq { + if (m.MsgTypeFlag & (MsgTypeFlagPositiveSeq | MsgTypeFlagNegativeSeq)) != 0 { writers = append(writers, m.writeSequence) }Apply the same bitmask logic in
readers()at Lines 425, 434:- if m.MsgTypeFlag == MsgTypeFlagPositiveSeq || m.MsgTypeFlag == MsgTypeFlagNegativeSeq { + if (m.MsgTypeFlag & (MsgTypeFlagPositiveSeq | MsgTypeFlagNegativeSeq)) != 0 { readers = append(readers, m.readSequence) } @@ - if m.MsgTypeFlag == MsgTypeFlagWithEvent { + if (m.MsgTypeFlag & MsgTypeFlagWithEvent) != 0 { readers = append(readers, m.readEvent, m.readSessionID, m.readConnectID) }
256-272: Major: String() equality checks break with combined flags.Lines 259, 266 use
==to check for sequence flags inString(), so messages with combined flags (event + sequence) won't display the sequence.Apply this diff:
- if m.MsgTypeFlag == MsgTypeFlagPositiveSeq || m.MsgTypeFlag == MsgTypeFlagNegativeSeq { + if (m.MsgTypeFlag & (MsgTypeFlagPositiveSeq | MsgTypeFlagNegativeSeq)) != 0 { return fmt.Sprintf("%s, %s, Sequence: %d, PayloadSize: %d", m.MsgType, m.EventType, m.Sequence, len(m.Payload)) } @@ - if m.MsgTypeFlag == MsgTypeFlagPositiveSeq || m.MsgTypeFlag == MsgTypeFlagNegativeSeq { + if (m.MsgTypeFlag & (MsgTypeFlagPositiveSeq | MsgTypeFlagNegativeSeq)) != 0 { return fmt.Sprintf("%s, %s, Sequence: %d, Payload: %s", m.MsgType, m.EventType, m.Sequence, string(m.Payload)) }
494-505: Major: Payload truncation not validated.Line 501 calls
buf.Next(int(size))but doesn't confirm the returned slice length matchessize. If the buffer is shorter,m.Payloadwill be truncated without error, causing silent data loss.Apply this diff to detect truncation:
if size > 0 { m.Payload = buf.Next(int(size)) + if uint32(len(m.Payload)) != size { + return fmt.Errorf("payload truncated: expected %d, got %d", size, len(m.Payload)) + } }relay/channel/volcengine/tts.go (1)
246-249: Major: Remove manual Transfer-Encoding header.Line 248 manually sets
"Transfer-Encoding: chunked", which conflicts with Go's net/http automatic chunking and can break HTTP/2 or proxies. TheFlush()calls at Line 281 are sufficient for streaming.Apply this diff:
contentType := getContentTypeByEncoding(encoding) c.Header("Content-Type", contentType) - c.Header("Transfer-Encoding", "chunked")
🧹 Nitpick comments (3)
relay/channel/volcengine/protocols.go (2)
459-462: Recommended: Validate SessionID length to catch truncation.Line 460 calls
buf.Next(int(size))without verifying the returned slice length. If the buffer is short, the session ID will be silently truncated.Apply this diff:
if size > 0 { m.SessionID = string(buf.Next(int(size))) + if uint32(len(m.SessionID)) != size { + return fmt.Errorf("session ID truncated: expected %d, got %d", size, len(m.SessionID)) + } }
479-482: Recommended: Validate ConnectID length to catch truncation.Line 480 calls
buf.Next(int(size))without checking the returned slice length, risking silent truncation.Apply this diff:
if size > 0 { m.ConnectID = string(buf.Next(int(size))) + if uint32(len(m.ConnectID)) != size { + return fmt.Errorf("connect ID truncated: expected %d, got %d", size, len(m.ConnectID)) + } }relay/channel/volcengine/tts.go (1)
298-305: Optional: Deduplicate usage object creation.Lines 286-291 and 299-303 create identical usage objects. Consider extracting this into a helper or setting it once after the loop.
For example:
func createTTSUsage(info *relaycommon.RelayInfo) *dto.Usage { return &dto.Usage{ PromptTokens: info.PromptTokens, CompletionTokens: 0, TotalTokens: info.PromptTokens, } }Then replace both instances with
usage = createTTSUsage(info).
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
relay/channel/volcengine/adaptor.go(8 hunks)relay/channel/volcengine/protocols.go(1 hunks)relay/channel/volcengine/tts.go(3 hunks)
🧰 Additional context used
🧬 Code graph analysis (2)
relay/channel/volcengine/tts.go (4)
relay/common/relay_info.go (1)
RelayInfo(75-122)types/error.go (7)
NewAPIError(87-95)NewErrorWithStatusCode(259-275)ErrorCodeChannelInvalidKey(57-57)ErrorCodeBadResponseStatusCode(70-70)ErrorCodeBadRequestBody(66-66)ErrorCodeBadResponse(71-71)ErrorCode(36-36)relay/channel/volcengine/protocols.go (6)
FullClientRequest(522-533)ReceiveMessage(507-520)MsgType(15-15)MsgTypeError(52-52)MsgTypeFrontEndResultServer(51-51)MsgTypeAudioOnlyServer(50-50)dto/openai_response.go (1)
Usage(222-235)
relay/channel/volcengine/adaptor.go (5)
constant/channel.go (2)
ChannelBaseURLs(60-117)ChannelTypeVolcEngine(45-45)relay/channel/api_request.go (1)
DoApiRequest(61-90)relay/common/relay_info.go (1)
RelayInfo(75-122)types/error.go (3)
NewAPIError(87-95)NewErrorWithStatusCode(259-275)ErrorCodeBadRequestBody(66-66)relay/channel/volcengine/tts.go (1)
VolcengineTTSRequest(21-26)
🔇 Additional comments (8)
relay/channel/volcengine/tts.go (5)
199-207: LGTM: Clean error handling for auth parsing.The function properly parses the API key and returns a well-structured error with the correct status code for authentication failures.
209-227: LGTM: Robust WebSocket dial with proper error handling.The dial logic correctly sets the Authorization header, handles both connection errors and HTTP response errors, and provides appropriate error codes for each case.
229-244: LGTM: Request marshaling and sending handled correctly.Proper error handling for JSON marshaling and WebSocket send operations with appropriate error codes.
250-261: LGTM: Receive loop with proper close detection.The loop correctly handles WebSocket close events (normal closure and going away) and returns appropriate errors for other failures.
263-295: LGTM: Message type handling and streaming logic are correct.The switch properly handles error messages, skips frontend result messages, streams audio chunks with flush, and terminates on negative sequence. Usage metrics are set correctly.
relay/channel/volcengine/adaptor.go (3)
26-29: LGTM: Context keys properly scoped.The unexported context keys follow Go conventions and avoid collisions.
260-264: LGTM: WebSocket URL correctly gated by official endpoint.Line 261 checks if the base URL matches the official Volcengine endpoint before returning the WebSocket URL, ensuring custom HTTP endpoints fall back to the HTTP path at Line 264.
312-326: LGTM: DoRequest bypasses HTTP for official streaming.Lines 313-324 correctly skip the HTTP request when streaming is enabled and the base URL is official, aligning with the WebSocket flow in
DoResponse.
…ream feat: doubao tts support streaming realtime audio
2 participants
豆包语音支持流式实时音频
Summary by CodeRabbit